package com.czk.springboot.reactive.reactor.subscribe;

import org.reactivestreams.Subscription;
import reactor.core.publisher.BaseSubscriber;
import reactor.core.publisher.Flux;
import reactor.core.publisher.SignalType;

public class OwnSubscribeDemo {

    public static void main(String[] args) {

        /**
         * doOnxx 发生这个事件后产生一个回调，通知你
         * onXXX 发生这个事件后执行一个动作，可以改变元素和信号
         *
         */

        Flux<String> flux = Flux.range(0, 5)
                .map(e -> {
                    System.out.println("转换");
                    return e + "czk";
                })
                // 流异常时，被捕获转为正常信号，订阅者不会感知
                .onErrorComplete()
                ;
        // 重写生命周期函数
        flux.subscribe(new BaseSubscriber<>() {
            // 1 订阅关系绑定是触发
            @Override
            protected void hookOnSubscribe(Subscription subscription) {
                System.out.println("订阅者绑定关系了");
                // 要一个数据
                super.request(1);
            }

            @Override
            protected void hookOnNext(String value) {
                System.out.println("数据到达");
                System.out.println(value);

                if (value.equals("czk")) {
                    // 取消流的订阅，订阅者随时可以取消
                    super.cancel();
                }


                // 要一个元素
                super.request(1);
            }

            @Override
            protected void hookOnComplete() {
                System.out.println("流完成了");
            }

            @Override
            protected void hookOnError(Throwable throwable) {
                System.out.println("流异常了");
            }

            @Override
            protected void hookOnCancel() {
                System.out.println("流取消了");
            }

            @Override
            protected void hookFinally(SignalType type) {
                super.hookFinally(type);
            }
        });

    }
}
